#include "base/logging.h"
#include "base/bind.h"
#include "base/thread_task_runner_handle.h"
#include "net/base/io_buffer.h"
#include "resource_dispatcher.h"
#include "resource_download.h"
#include "resource_storage.h"
#include "common/defines.h"

namespace dl {
using namespace net;

namespace {
MSG_Finished_Param ResolveErrorCode(const URLRequestStatus& stat) {
  MSG_Finished_Param param;
  switch (stat.status()) {
  case URLRequestStatus::CANCELED:
    param.error_code = kDownloadCancel;
    param.error_desc = L"取消下载";
    break;
  case URLRequestStatus::FAILED:
    param.error_code = kDownloadFailed;
    param.error_desc = L"下载失败";
    break;
  default:
    param.error_code = kDownloadSuccess;
    param.error_desc = L"下载成功";
    break;
  }
  return param;
}
}

// ResourceDispatcher::RequestContext
// static
ResourceDispatcher::RequestContext* ResourceDispatcher::RequestContext::Create(
    WorkChannel* c, URLRequestData data) {
  ResourceDispatcher::RequestContext* context = new ResourceDispatcher::RequestContext;
  context->channels.push_back(c->AsWeakPtr());
  context->request_data = data;
  context->content_length = std::make_pair((int64)0, (int64)0);
  return context;
}

// ResourceDispatcher::ResponseHandler
ResourceDispatcher::ResponseHandler::ResponseHandler(ResourceDispatcher& d)
    : dispatcher(d){

}

// ResourceDownload::ResponseDelegate接口, 运行在download_thread_线程
bool ResourceDispatcher::ResponseHandler::OnResponseStarted(
    unsigned int symbol, ResponsePtr response) {
  dispatcher.main_task_runner_->PostTask(FROM_HERE,
      base::Bind(&ResourceDispatcher::OnResponseStarted,
                 base::Unretained(&dispatcher), symbol, response));
  return true;
}

void ResourceDispatcher::ResponseHandler::OnReadCompleted(
    unsigned int symbol, IOBufferWithSizePtr buffer) {
  dispatcher.main_task_runner_->PostTask(FROM_HERE,
      base::Bind(&ResourceDispatcher::OnReadCompleted,
                 base::Unretained(&dispatcher), symbol, buffer));
}

void ResourceDispatcher::ResponseHandler::OnResponseFinished(
    unsigned int symbol, net::URLRequestStatus stat) {
  dispatcher.main_task_runner_->PostTask(FROM_HERE,
      base::Bind(&ResourceDispatcher::OnResponseFinished,
                 base::Unretained(&dispatcher), symbol, stat));
}

// ResourceDispatcher::StorageHandler
ResourceDispatcher::StorageHandler::StorageHandler(ResourceDispatcher& d)
: dispatcher(d){

}

// ResourceStorage::StorageDelegate接口, 运行在file_thread_线程
void ResourceDispatcher::StorageHandler::OnCreateStorageCompleted(
    unsigned int symbol, int64 start_position) {
  dispatcher.main_task_runner_->PostTask(FROM_HERE,
      base::Bind(&ResourceDispatcher::OnCreateStorageCompleted,
                 base::Unretained(&dispatcher), symbol, start_position));
}


// ResourceDispatcher
ResourceDispatcher::ResourceDispatcher() :
    main_task_runner_(base::ThreadTaskRunnerHandle::Get()) {
  DCHECK(MessageLoop::current()->type() == MessageLoop::TYPE_UI);
  response_handler_.reset(new ResponseHandler(*this));
  storage_handler_.reset(new StorageHandler(*this));
  CreateDownloadThread();
  CreateFileThread();
}

ResourceDispatcher::~ResourceDispatcher() {
  file_thread_->Stop();
  download_thread_->Stop();
}

void ResourceDispatcher::Clear() {
  RequestContextMap::iterator iter = request_context_.begin();
  for ( ; iter != request_context_.end(); ++iter) {
    download_thread_->PostTask(FROM_HERE,
        base::Bind(&ResourceDownload::CancelUrlRequest,
                   base::Unretained(download_.get()), iter->first));
  }
}

bool ResourceDispatcher::CreateDownloadThread() {
  // 创建专门用于URL下载的线程
  download_thread_.reset(new WorkThread("DownloadThread", this));
  base::Thread::Options options;
  options.message_loop_type = MessageLoop::TYPE_IO;
  return download_thread_->StartWithOptions(options);
}

bool ResourceDispatcher::CreateFileThread() {
  // 创建专门用于文件写入的线程
  file_thread_.reset(new WorkThread("FileThread", this));
  base::Thread::Options options;
  options.message_loop_type = MessageLoop::TYPE_IO;
  return file_thread_->StartWithOptions(options);
}

// MessageTransfer 接口
bool ResourceDispatcher::OnMessageDispatch(
    WorkChannel* channel, const IPC::Message& message) {
  bool handled = true;
  IPC_BEGIN_MESSAGE_MAP(ResourceDispatcher, message)
    IPC_MESSAGE_HANDLER_WITH_SENDER(channel, MSG_Request, OnRequestDownload)
    IPC_MESSAGE_HANDLER_WITH_SENDER(channel, MSG_Cancel, OnCancelDownload)
    IPC_MESSAGE_UNHANDLED(handled = false)
  IPC_END_MESSAGE_MAP()
  return handled;
}

// IPC消息处理
void ResourceDispatcher::OnRequestDownload(
    WorkChannel* channel, unsigned int symbol, const MSG_Request_Param& param) {
  DCHECK(CalledOnValidThread());
	
  RequestContextMap::iterator iter = request_context_.find(symbol);
  if (iter == request_context_.end()) {
    URLRequestData data;
    data.request_url      =  GURL(param.request_url.value());
    data.save_path        =  param.save_path;
    data.start_position   =  param.start_position;
    RequestContextPtr context(RequestContext::Create(channel, data));
    request_context_[symbol] = context;
    // 准备存储位置
    StorageCreateInfo info;
    info.save_path = context->request_data.save_path;
    info.start_position = context->request_data.start_position;
    file_thread_->PostTask(FROM_HERE,
        base::Bind(&ResourceStorage::CreateStorage,
                   base::Unretained(storage_.get()), symbol, info));
  } else {
    RequestContextPtr& context = iter->second;
    context->channels.push_back(channel->AsWeakPtr());
  }
}

void ResourceDispatcher::OnCancelDownload(WorkChannel* channel, unsigned int symbol) {
  download_thread_->PostTask(FROM_HERE,
      base::Bind(&ResourceDownload::CancelUrlRequest,
                 base::Unretained(download_.get()), symbol));
}

void ResourceDispatcher::OnResponseStarted(unsigned int symbol, ResponsePtr response) {
  DCHECK(CalledOnValidThread());
  RequestContextMap::iterator iter = request_context_.find(symbol);
  if (iter != request_context_.end()) {
    RequestContextPtr& context = iter->second;
    context->content_length.first  = response->start_position;
    context->content_length.second = response->content_length;
    // 定位文件存储位置
    file_thread_->PostTask(FROM_HERE, 
        base::Bind(&ResourceStorage::PositionStorage,
                   base::Unretained(storage_.get()), symbol, response->start_position));
    // 开始统计速度
    context->speed_statistics.Start(
        base::Bind(&ResourceDispatcher::OnNotifyProgress,
                   base::Unretained(this), symbol, context));
    // 给连接端发送回复通知
    WorkChannelVector::iterator chan_iter = context->channels.begin();
    for (; chan_iter != context->channels.end(); ++chan_iter) {
      base::WeakPtr<WorkChannel>& channel = *chan_iter;
      if (channel) {
        MSG_Response_Param param;
        param.response_code = response->headers->response_code();
        channel->Send(new MSG_Response(MSG_ROUTING_NONE, symbol, param));
      }
    }
  }
}

void ResourceDispatcher::OnReadCompleted(unsigned int symbol, IOBufferWithSizePtr buffer) {
  DCHECK(CalledOnValidThread());
  // 将读取到的数据写入文件
  file_thread_->PostTask(FROM_HERE, 
      base::Bind(&ResourceStorage::WriteToStorage,
                 base::Unretained(storage_.get()), symbol, buffer));
  // 采样速度统计数据
  RequestContextMap::iterator iter = request_context_.find(symbol);
  if (iter != request_context_.end()) {
    RequestContextPtr& context = iter->second;
    context->content_length.first += buffer->size();
    context->speed_statistics.Record(buffer->size());
  }
}

void ResourceDispatcher::OnResponseFinished(unsigned int symbol, net::URLRequestStatus stat) {
  DCHECK(CalledOnValidThread());
  // 关闭文件写入流程
  file_thread_->PostTask(FROM_HERE, 
      base::Bind(&ResourceStorage::StorageFinished,
                 base::Unretained(storage_.get()), symbol, stat));
  // 给相关管道发送结束通知
  RequestContextMap::iterator iter = request_context_.find(symbol);
  if (iter != request_context_.end()) {
    RequestContextPtr& context = iter->second;
    context->speed_statistics.Stop();
    OnNotifyProgress(symbol, context);
    WorkChannelVector::iterator chan_iter;
    for (chan_iter = context->channels.begin(); 
        chan_iter != context->channels.end(); ++chan_iter) {
      base::WeakPtr<WorkChannel>& channel = *chan_iter;
      if (channel) {
        channel->Send(new MSG_Finished(MSG_ROUTING_NONE, symbol, ResolveErrorCode(stat)));
      }
    }
    request_context_.erase(iter);
  }
}

void ResourceDispatcher::OnCreateStorageCompleted(unsigned int symbol, int64 start_position) {
  RequestContextMap::iterator iter = request_context_.find(symbol);
  if (iter != request_context_.end()) {
    RequestContextPtr& context = iter->second;
    download_thread_->PostTask(FROM_HERE,
        base::Bind(&ResourceDownload::AddUrlRequest,
                   base::Unretained(download_.get()), symbol, context->request_data));
  }
}

void ResourceDispatcher::OnNotifyProgress(unsigned int symbol, RequestContextPtr context) {
  WorkChannelVector::iterator chan_iter;
  for (chan_iter = context->channels.begin(); 
      chan_iter != context->channels.end(); ++chan_iter) {
    base::WeakPtr<WorkChannel>& channel = *chan_iter;
    if (channel) {
      MSG_Progress_Param param;
      param.already_bytes = context->content_length.first;
      param.total_bytes   = context->content_length.second;
      param.bytes_per_sec = context->speed_statistics.CurrentSpeed();
      channel->Send(new MSG_Progress(MSG_ROUTING_NONE, symbol, param));
    }
  }
}

// WorkThread::Delegate接口
void ResourceDispatcher::OnInit(WorkThread* thr) {
  if (thr == download_thread_.get()) {
    download_.reset(new ResourceDownload(response_handler_.get()));  
  } else if (thr == file_thread_.get()) {
    storage_.reset(new ResourceStorage(storage_handler_.get()));
  }
}

void ResourceDispatcher::OnCleanup(WorkThread* thr) {
  if (thr == download_thread_.get()) {
    download_.reset();
  } else if (thr == file_thread_.get()) {
    storage_.reset();
  }
}

}
